#include "subscribe_manager.h"
#include "client_session.h"

namespace acl
{
namespace mqtt
{
	subscribe_manager & subscribe_manager::get_instance ()
	{
		static subscribe_manager instance;
		return instance;
	}

	void subscribe_manager::pub_msg (publish_msg &&msg)
	{
		auto &clisess = topic_map_[msg.topic_name_];
		if (clisess.empty ())
			return;
		for (auto itr = clisess.begin (); itr != clisess.end ();) {
			auto sess = itr->second.client_session_.lock ();
			if (!sess) {//remove dead sess
				itr = clisess.erase (itr);
				continue;
			}
			sess->recv_msg (msg, false);
			++itr;
		}
	}

	bool subscribe_manager::sub (std::shared_ptr<client_session> cli_sess, 
								 subscribe_msg::topic_filter topic_filter)
	{
		if (cli_sess->get_id ().empty () ||
			topic_filter.topic_.empty ())
			return false;

		topic_map_[topic_filter.topic_][cli_sess->get_id ()]
			= {topic_filter,cli_sess};

		auto &lmsg = msg_storer::get_instance ().get_msgs (topic_filter.topic_);
		if (lmsg.empty ())
			return true;
		for (auto &itr : lmsg)
			cli_sess->recv_msg (itr, true);

		return true;
	}

	bool subscribe_manager::unsub (const std::string &topic, 
								   const std::string &session_id)
	{
		auto &clisess = topic_map_[topic];
		if (clisess.empty ())
			return false;
		auto itr = clisess.find (session_id);
		if (itr != clisess.end ()) {
			clisess.erase (itr);
			return true;
		}
		return false;
	}

	subscribe_manager::subscribe_manager ()
	{

	}
}
}